package com.lagou.client;

import com.alibaba.fastjson.JSONObject;
import com.lagou.balance.Balance;
import com.lagou.balance.ShortTimeBalance;
import com.lagou.handler.StringChannelHandler;
import com.lagou.handler.UserClientHandler;
import com.lagou.request.RpcRequest;
import com.lagou.stat.RequestInfo;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 消费者
 */
public class RPCConsumer {

    // 1.创建一个线程池对象 -- 它要处理我们自定义事件
    private static ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 3. 自定义事件处理器
    private static final Map<String, List<UserClientHandler>> handlerMap = new ConcurrentHashMap<>();

    // 数据上报
    private static final Map<String, Map<String, RequestInfo>> statMap = new ConcurrentHashMap<>();

    private static CuratorFramework curatorFramework;

    private static Balance balance;

    private static final String SERVICE = "/services";
    private static final String STAT = "/stat";

    // 加载注册的服务
    public static void loadRegisterService(String serviceName) throws Exception {
        // 1. 初始化客户端
        RetryPolicy policy = new ExponentialBackoffRetry(1000, 10);
        curatorFramework = CuratorFrameworkFactory
                .builder()
                .connectString("127.0.0.1:2181")
                .sessionTimeoutMs(10000)
                .retryPolicy(policy)
                .namespace("base")
                .build();
        curatorFramework.start();
        // 2. 加载子节点,并初始化客户端
        String path = String.format("%s/%s", SERVICE, serviceName);
        List<String> serverList = curatorFramework.getChildren().forPath(path);
        for (String server : serverList) {
            String[] address = server.split(":");
            // 1. 初始化客户端
            initClient(serviceName, address[0], Integer.parseInt(address[1]));
            //3. 初始化上报数据节点
            registerStatMap(serviceName, server);
        }
        //2. 添加监听
        addListener(serviceName);
        // 4. 注册定时器
        scheduleStatMap(serviceName);
        // 5. 初始化负载据均衡
        balance = new ShortTimeBalance(curatorFramework);
    }

    public static void scheduleStatMap(String serviceName) {
        Timer timer = new Timer();
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                Map<String, RequestInfo> requestInfoMap = statMap.get(serviceName);
                for (Map.Entry<String, RequestInfo> entry : requestInfoMap.entrySet()) {
                    String address = entry.getKey();
                    String path = String.format("%s/%s/%s", STAT, serviceName, address);
                    RequestInfo requestInfo = entry.getValue();
                    long lastRequestTime = requestInfo.getLastRequestTime();
                    // 5秒没有请求
                    if (lastRequestTime + 5_000L < System.currentTimeMillis()) {
                        requestInfo.setLastRequestTime(0L);
                        requestInfo.setLastUseTime(0L);
                    }
                    try {
                        curatorFramework.setData().forPath(path, JSONObject.toJSONBytes(requestInfo));
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }, 5000L, 5000L);
    }

    public static void registerStatMap(String serviceName, String server) throws Exception {

        RequestInfo requestInfo = new RequestInfo();

        Map<String, RequestInfo> requestInfoMap = new ConcurrentHashMap<>();
        requestInfoMap.put(server, requestInfo);

        statMap.put(serviceName, requestInfoMap);

        String statPath = String.format("%s/%s/%s", STAT, serviceName, server);
        Stat stat = curatorFramework.checkExists().forPath(statPath);
        if (stat == null) {
            curatorFramework.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(statPath, JSONObject.toJSONBytes(requestInfo));
        }
    }

    public static void reportStatMap(String serviceName, String server, Long userTime) throws Exception {
        Map<String, RequestInfo> requestInfoMap = statMap.getOrDefault(serviceName, new ConcurrentHashMap<>());
        RequestInfo requestInfo = requestInfoMap.get(server);
        if (requestInfo == null) {
            requestInfo = new RequestInfo();
        }
        requestInfo.setLastRequestTime(System.currentTimeMillis());
        requestInfo.setLastUseTime(userTime);
        requestInfoMap.put(server, requestInfo);
        statMap.put(serviceName, requestInfoMap);
    }

    public static void addListener(String serviceName) throws Exception {
        String servicePath = String.format("%s/%s", SERVICE, serviceName);
        PathChildrenCache nodeCache = new PathChildrenCache(curatorFramework, servicePath, true);
        nodeCache.start();
        nodeCache.getListenable().addListener(new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent pathChildrenCacheEvent) throws Exception {
                if (pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CHILD_REMOVED) {
                    String path = pathChildrenCacheEvent.getData().getPath();
                    List<UserClientHandler> handlers = handlerMap.get(serviceName);
                    int index = path.lastIndexOf("/");
                    path = path.substring(index + 1);
                    for (UserClientHandler handler : handlers) {
                        if (handler.getAddress().equals(path)) {
                            handlers.remove(handler);
                            break;
                        }
                    }
                } else if (pathChildrenCacheEvent.getType() == PathChildrenCacheEvent.Type.CHILD_ADDED) {
                    System.out.println("增加了客户端。。。。。。。。");
                    String path = pathChildrenCacheEvent.getData().getPath();
                    int index = path.lastIndexOf("/");
                    path = path.substring(index + 1);
                    List<UserClientHandler> handlers = handlerMap.get(serviceName);
                    if (handlers == null) {
                        handlers = new ArrayList<>();
                    }
                    boolean exist = false;
                    for (UserClientHandler userClientHandler : handlers) {
                        if (userClientHandler.getAddress().equals(path)) {
                            exist = true;
                            break;
                        }
                    }
                    if (!exist) {
                        String[] service = path.split(":");
                        initClient(serviceName, service[0], Integer.parseInt(service[1]));
                    }
                }
            }
        });
    }

    // 3.编写方法,初始化客户端 ( 创建连接池 bootStrap 设置bootstrap 连接服务器)
    public static void initClient(String serviceName, String host, int port) throws InterruptedException {
        // 1) 初始化UserClientHandler
        UserClientHandler userClientHandler = new UserClientHandler();
        userClientHandler.setAddress(String.format("%s:%d", host, port));
        // 2)创建连接池对象
        EventLoopGroup group = new NioEventLoopGroup();
        // 3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();
        // 4)配置启动引导对象
        bootstrap.group(group)
                // 设置通道为NIO
                .channel(NioSocketChannel.class)
                // 设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                // 监听channel 并初始化
                .handler(new StringChannelHandler(userClientHandler));

        // 5)连接服务端
        bootstrap.connect(host, port).sync();
        List<UserClientHandler> handlers = handlerMap.getOrDefault(serviceName, new ArrayList<>());
        handlers.add(userClientHandler);
        handlerMap.put(serviceName, handlers);
    }

    // 4.编写一个方法,使用JDK的动态代理创建对象
    // serviceClass 接口类型,根据哪个接口生成子类代理对象; providerParam : "UserService#sayHello#"
    public static Object createProxy(Class<?> serviceClass, final RpcRequest rpcRequest) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class[]{serviceClass},
                new InvocationHandler() {
                    public Object invoke(Object o, Method method, Object[] objects) throws Throwable {
                        String serviceName = serviceClass.getSimpleName();
                        // 1)初始化客户端cliet
                        List<UserClientHandler> handlers = handlerMap.get(serviceName);
                        if (handlers == null) {
                            loadRegisterService(serviceName);
                        }
                        // 2)给UserClientHandler 设置param参数
                        UserClientHandler userClientHandler = selectHandler(serviceName);
                        userClientHandler.setRpcRequest(rpcRequest);

                        long begTime = System.currentTimeMillis();
                        // 3).使用线程池,开启一个线程处理处理call() 写操作,并返回结果
                        Object result = executorService.submit(userClientHandler).get();
                        long useTime = System.currentTimeMillis() - begTime;
                        reportStatMap(serviceName, userClientHandler.getAddress(), useTime);
                        // 4)return 结果
                        return String.format("%s【%s】", result, userClientHandler.getAddress());
                    }
                });
    }

    private static UserClientHandler selectHandler(String serviceName) throws Exception {
        List<UserClientHandler> list = handlerMap.get(serviceName);
        return balance.choose(list);
    }

}
